package com.example.rxjavalearn2.simple_rxjava;

/**
 * Description: 创建Observable的类
 * Author: wucan
 * Date: 20-5-29 下午4:18
 */
public class ObservableCreat<T> extends Obserable<T>{

    ObservableOnSubscribe<T> source;

    public ObservableCreat(ObservableOnSubscribe<T> source) {
        this.source = source;
    }


    @Override
    public void subscirbeActual(Observer<T> observer) {
        System.out.println("ObservableCreat."+"subscirbeActual");
        //创建发射器
        MEmitter<T> emitter = new CreateEmitter(observer);
        //ObservableOnSubscribe的实现类调用
        source.subscribe(emitter);
    }


    class CreateEmitter implements MEmitter<T>{


        Observer<T> observer;

        public CreateEmitter(Observer<T> observer) {
            this.observer = observer;
        }

        @Override
        public void onNext(T value) {
            System.out.println("CreateEmitter.onNext");
            observer.onNext(value);
        }

        @Override
        public void onError(Throwable error) {
            observer.onError(error);
        }

        @Override
        public void onComplete() {
            observer.onComplete();
        }
    }
}
